package com.cxdr.awsips.distribute.processor;

import java.util.HashMap;
import java.util.Map;

import org.apache.log4j.Logger;

import com.cxdr.awsips.common.logger.LogFactory;
import com.cxdr.awsips.common.thread.BaseProcessor;
import com.cxdr.awsips.distribute.bean.Link_Target;
import com.cxdr.awsips.distribute.bean.ReportFileLog;
import com.cxdr.awsips.distribute.config.LinkConfig;
import com.cxdr.awsips.distribute.pool.BaseQueue;
/**
 * 分发文件调度器
 * 向各个文件分发器分发文件
 * @author pugang
 *
 */
public class OutFileDispatcher extends BaseProcessor {
	private static Logger logger = LogFactory.makeNewLogger(OutFileDispatcher.class);
	public static final String THREAD_NAME = "OutFileDispatcher";
	public  final BaseQueue<ReportFileLog> OutPutFileQueue = new BaseQueue<ReportFileLog>();
	private Map<String,DistributeProccessor> proccessorMap = new HashMap<String,DistributeProccessor>();
//	private ExecutorService executor;
	private LinkConfig link = null;
	public OutFileDispatcher(LinkConfig link){
		this.link = link;
		setTask_name(DistributeWorker.TASK_NAME);
		setThread_name(THREAD_NAME);
//		executor = Executors.newFixedThreadPool(link.getTargetSize());
		for(Link_Target t:link.getTargets()){
			DistributeProccessor proccessor = new DistributeProccessor(link,t);
			proccessorMap.put(t.getId(), proccessor);
//			executor.submit(proccessor);
		}
	}
	@Override
	@Deprecated
	public void run(){
		logger.info("开始OutFileDispatcher");
		while(!isFinish()){
			ReportFileLog file = OutPutFileQueue.take();
			if(file==null){
				logger.info("文件是空");
				continue;
			}
			notifiAllProccessor(file);
		}
		logger.info("OutFileDispatcher finished");
	}
	
	public void beginProccess(){
		while(!OutPutFileQueue.isEmpty()){
			ReportFileLog fileLog = OutPutFileQueue.take();
			notifiAllProccessor(fileLog);
		}
		for(DistributeProccessor proccessor:proccessorMap.values()){
			proccessor.distribute();
		}
	}
	//通知所有分发处理器
	private void notifiAllProccessor(ReportFileLog fileLog){
		logger.info("通知文件 "+fileLog.getFile().getName());
		for(DistributeProccessor proccessor:proccessorMap.values()){
			if(proccessor.accept(fileLog.getFile())){
				proccessor.DistributeFileQueue.add(fileLog.getFile());
			}
		}
	}
	@Deprecated
	private void shutdownAllProcessor(){
		for(DistributeProccessor proccessor:proccessorMap.values()){
			proccessor.finish();
		}
	}
}
